package trouger.dmt;

import java.util.HashMap;
import java.util.HashSet;

/**
 * Some predefined aggregation operators.
 * Common applications will find these operators sufficient.
 */
public class PredefinedAggregations {
	/**
	 * Compute the sum of integers, and give an integer as the result.
	 * Both incoming integers and the result integer must fit in 32-bit Integers.
	 */
	public static class IntegerSum
	implements AggregationOperator{
		int sum = 0;
		boolean hasValue = false;
		@Override
		public void acceptValue(String value) {
			sum += Integer.parseInt(value);
			hasValue = true;
		}
		@Override
		public String getAggregatedValue() {
			return hasValue ? new Integer(sum).toString() : null;
		}
	}
	/**
	 * Compute the sum of integers, and give an integer as the result.
	 * Both incoming integers and the result integer must fit in 64-bit Integers.
	 */
	public static class LongintSum
	implements AggregationOperator{
		long sum = 0;
		boolean hasValue = false;
		@Override
		public void acceptValue(String value) {
			sum += Long.parseLong(value);
			hasValue = true;
		}
		@Override
		public String getAggregatedValue() {
			return hasValue ? new Long(sum).toString() : null;
		}
	}
	/**
	 * Compute the sum of real numbers, and give a real number as the result.
	 * Double precision floating point arithmetic is used in this class.
	 */
	public static class FloatSum
	implements AggregationOperator{
		double sum = 0;
		boolean hasValue = false;
		@Override
		public void acceptValue(String value) {
			sum += Double.parseDouble(value);
			hasValue = true;
		}
		@Override
		public String getAggregatedValue() {
			return hasValue ? new Double(sum).toString() : null;
		}
	}
	/**
	 * Compute the average of a sequence of real numbers.
	 * Double precision floating point arithmetic is used in this class.
	 */
	public static class Average
	implements AggregationOperator{
		double sum = 0;
		double count = 0;
		@Override
		public void acceptValue(String value) {
			sum += Double.parseDouble(value);
			count++;
		}
		@Override
		public String getAggregatedValue() {
			if (count == 0){
				return null;
			}else{
				return new Double(sum / count).toString();
			}
		}
	}
	/**
	 * Compute the variance of a sequence of real numbers.
	 * Double precision floating point arithmetic is used in this class.
	 */
	public static class Variance
	implements AggregationOperator{
		double sum = 0;
		double sqrsum = 0;
		double count = 0;
		@Override
		public void acceptValue(String value) {
			double t = Double.parseDouble(value);
			sum += t;
			sqrsum += t * t;
			count++;
		}
		@Override
		public String getAggregatedValue() {
			if (count == 0){
				return null;
			}else{
				double t;
				t = sqrsum / count - (t = sum / count) * t;
				return new Double(t).toString();
			}
		}
	}
	/**
	 * Compute the standard deviation of a sequence of real numbers.
	 * Double precision floating point arithmetic is used in this class.
	 */
	public static class StandardDeviation
	implements AggregationOperator{
		double sum = 0;
		double sqrsum = 0;
		double count = 0;
		@Override
		public void acceptValue(String value) {
			double t = Double.parseDouble(value);
			sum += t;
			sqrsum += t * t;
			count++;
		}
		@Override
		public String getAggregatedValue() {
			if (count == 0){
				return null;
			}else{
				double t;
				t = sqrsum / count - (t = sum / count) * t;
				t = Math.sqrt(t);
				return new Double(t).toString();
			}
		}
	}
	/**
	 * Compute the number of records.
	 * The aggregated value is exactly the number of times acceptValue is called.
	 */
	public static class Count
	implements AggregationOperator{
		long count = 0;
		@Override
		public void acceptValue(String value) {
			count++;
		}
		@Override
		public String getAggregatedValue() {
			return new Long(count).toString();
		}
	}
	/**
	 * Compute the number of different values in a list of String objects.
	 * The aggregated value is exactly the cardinality of the set of all strings.
	 */
	public static class UniqueCount
	implements AggregationOperator{
		HashSet<String> set = new HashSet<String>();
		@Override
		public void acceptValue(String value) {
			set.add(value);
		}
		@Override
		public String getAggregatedValue() {
			return new Integer(set.size()).toString();
		}
	}
	/**
	 * Compute the maximum value of 32-bit integers.
	 */
	public static class IntegerMax
	implements AggregationOperator{
		boolean firstTime = true;
		int max = 0;
		@Override
		public void acceptValue(String value) {
			if (firstTime){
				max = Integer.parseInt(value);
				firstTime = false;
			}else{
				int t = Integer.parseInt(value);
				if (t > max) max = t;
			}
		}
		@Override
		public String getAggregatedValue() {
			if (firstTime){
				return null;
			}else{
				return new Integer(max).toString();
			}
		}
	}
	/**
	 * Compute the minimum value of 32-bit integers.
	 */
	public static class IntegerMin
	implements AggregationOperator{
		boolean firstTime = true;
		int min = 0;
		@Override
		public void acceptValue(String value) {
			if (firstTime){
				min = Integer.parseInt(value);
				firstTime = false;
			}else{
				int t = Integer.parseInt(value);
				if (t < min) min = t;
			}
		}
		@Override
		public String getAggregatedValue() {
			if (firstTime){
				return null;
			}else{
				return new Integer(min).toString();
			}
		}
	}
	/**
	 * Compute the maximum value of 64-bit integers.
	 */
	public static class LongintMax
	implements AggregationOperator{
		boolean firstTime = true;
		long max = 0;
		@Override
		public void acceptValue(String value) {
			if (firstTime){
				max = Long.parseLong(value);
				firstTime = false;
			}else{
				long t = Long.parseLong(value);
				if (t > max) max = t;
			}
		}
		@Override
		public String getAggregatedValue() {
			if (firstTime){
				return null;
			}else{
				return new Long(max).toString();
			}
		}
	}
	/**
	 * Compute the minimum value of 64-bit integers.
	 */
	public static class LongintMin
	implements AggregationOperator{
		boolean firstTime = true;
		long min = 0;
		@Override
		public void acceptValue(String value) {
			if (firstTime){
				min = Long.parseLong(value);
				firstTime = false;
			}else{
				long t = Long.parseLong(value);
				if (t < min) min = t;
			}
		}
		@Override
		public String getAggregatedValue() {
			if (firstTime){
				return null;
			}else{
				return new Long(min).toString();
			}
		}
	}
	/**
	 * Compute the maximum value of 64-bit doubles.
	 */
	public static class FloatMax
	implements AggregationOperator{
		boolean firstTime = true;
		double max = 0;
		@Override
		public void acceptValue(String value) {
			if (firstTime){
				max = Double.parseDouble(value);
				firstTime = false;
			}else{
				double t = Double.parseDouble(value);
				if (t > max) max = t;
			}
		}
		@Override
		public String getAggregatedValue() {
			if (firstTime){
				return null;
			}else{
				return new Double(max).toString();
			}
		}
	}
	/**
	 * Compute the minimum value of 64-bit doubles.
	 */
	public static class FloatMin
	implements AggregationOperator{
		boolean firstTime = true;
		double min = 0;
		@Override
		public void acceptValue(String value) {
			if (firstTime){
				min = Double.parseDouble(value);
				firstTime = false;
			}else{
				double t = Double.parseDouble(value);
				if (t < min) min = t;
			}
		}
		@Override
		public String getAggregatedValue() {
			if (firstTime){
				return null;
			}else{
				return new Double(min).toString();
			}
		}
	}
	/**
	 * The aggregated value is the value that comes first.
	 */
	public static class FirstValue
	implements AggregationOperator{
		boolean firstTime = true;
		String value = null;
		@Override
		public void acceptValue(String value) {
			if (firstTime){
				this.value = value;
				firstTime = false;
			}
		}
		@Override
		public String getAggregatedValue() {
			return value;
		}
	}
	/**
	 * The aggregated value is the last incoming value.
	 */
	public static class LastValue
	implements AggregationOperator{
		String value = null;
		@Override
		public void acceptValue(String value) {
			this.value = value;
		}
		@Override
		public String getAggregatedValue() {
			return value;
		}
	}
	
	protected static HashMap<String, Class<? extends AggregationOperator>> aliasMap;
	static{
		aliasMap = new HashMap<String, Class<? extends AggregationOperator>>();
		aliasMap.put("sum", FloatSum.class);
		aliasMap.put("sumi", IntegerSum.class);
		aliasMap.put("suml", LongintSum.class);
		aliasMap.put("sumf", FloatSum.class);
		aliasMap.put("avr", Average.class);
		aliasMap.put("var", Variance.class);
		aliasMap.put("dev", StandardDeviation.class);
		aliasMap.put("count", Count.class);
		aliasMap.put("count_distinct", UniqueCount.class);
		aliasMap.put("max", FloatMax.class);
		aliasMap.put("maxi", IntegerMax.class);
		aliasMap.put("maxl", LongintMax.class);
		aliasMap.put("maxf", FloatMax.class);
		aliasMap.put("min", FloatMin.class);
		aliasMap.put("mini", IntegerMin.class);
		aliasMap.put("minl", LongintMin.class);
		aliasMap.put("minf", FloatMin.class);
		aliasMap.put("first", FirstValue.class);
		aliasMap.put("last", LastValue.class);
	}
	/**
	 * Get a predefined aggregation operator class by its alias.
	 * @param alias the name of the aggregation operator.
	 * @return a predefined AggregationOperator implementation class.
	 */
	public static Class<? extends AggregationOperator> getOperatorClassByAlias(String alias){
		return aliasMap.get(alias.toLowerCase());
	}
}
